package com.surfilter.massdata.spark.task.daystat2;

import java.util.ArrayList;
import java.util.Date;
import java.util.HashMap;
import java.util.Iterator;
import java.util.LinkedList;
import java.util.List;
import java.util.Map;

import org.apache.commons.lang3.StringUtils;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.JavaSparkContext;
import org.apache.spark.api.java.function.FlatMapFunction;
import org.apache.spark.api.java.function.Function;
import org.apache.spark.broadcast.Broadcast;
import org.apache.spark.sql.DataFrame;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SQLContext;
import org.apache.spark.storage.StorageLevel;

import com.act.sparkanalyz.log.SysLog;
import com.act.sparkanalyz.service.impl.SparkService.OutQueueEntity;
import com.act.sparkanalyz.task.ISparkTask;
import com.surfilter.massdata.spark.bean.ActiveIpBean;
import com.surfilter.massdata.spark.bean.SAN051;
import com.surfilter.massdata.spark.model.DnsStatConsts;
import com.surfilter.massdata.spark.model.StructIP;
import com.surfilter.massdata.spark.util.CommonUtils;
import com.surfilter.massdata.spark.util.DateUtil;
import com.surfilter.massdata.spark.util.IPDatabase;

public class ActiveIPAnalyzTask_New implements ISparkTask {
	private static final long serialVersionUID = 1L;
	private String city_top;
	private String oper_top;

	@Override
	public List<OutQueueEntity> execute(Map<String, DataFrame> dataFrames,Map<String, String> commandMap) {
		long start=System.currentTimeMillis();
		List<OutQueueEntity> list = new ArrayList<OutQueueEntity>();
		try{
			DataFrame cipFilterFrame = dataFrames.get("dws_hour_san051");
			SQLContext sqlContext = cipFilterFrame.sqlContext();
			
			DataFrame operators_id_param=sqlContext.sql("select distinct OPERATORS_ID as operators_id from br2002_temp");
			operators_id_param.registerTempTable("operators_id_param");
			
			DataFrame city_param = sqlContext.sql("select distinct CITY as city from br2002_temp");
			city_param.registerTempTable("city_param");
	
			String dayStr = commandMap.get("-d");
			Date date = DateUtil.getExecDate(new Date(), dayStr);
			// 活跃IP排名-地市用户IP访问量排名（50）
			DataFrame city_top = getCityTopDF(city_param, date);
			list.add(new OutQueueEntity(this.city_top, city_top));

			// 活跃IP排名-运营商用户IP访问量排名（50）
			DataFrame oper_top = getOperatorTopDF(operators_id_param, date);
			list.add(new OutQueueEntity(this.oper_top, oper_top));
			
			//System.out.println("city_top:"+city_top.count());
			//System.out.println("oper_top:"+oper_top.count());
			
			sqlContext.dropTempTable("operators_id_param");
			sqlContext.dropTempTable("city_param");
			CommonUtils.deleteTaskTableData("SAN051", date, "WEBSITE_COUNT_AREA", 0, "day");
			CommonUtils.deleteTaskTableData("SAN051", date, "WEBSITE_COUNT_OPERATORS", 0, "day");
		}
		catch(Exception e){
			SysLog.error(e.getMessage());
		}
		
		long end=System.currentTimeMillis();
		double min=(end-start)*1.0/(1000*60);
		System.out.println("ActiveIPAnalyzTask_New:exectime: "+min+" min............");
		
		return list;
	}

	@SuppressWarnings("serial")
	private DataFrame getOperatorTopDF(DataFrame operators_id_param, final Date date) {
		
		String sql = "select b.buss_value,b.vcount,b.client_ip,b.rn from( "+
				"select a.buss_value,a.client_ip,a.vcount,ROW_NUMBER() OVER(PARTITION BY a.buss_value ORDER BY a.vcount DESC) rn from( "+
				"select buss_value,client_ip,sum(count_value)as vcount from dws_hour_san051_temp  where buss_type='WEBSITE_COUNT_OPERATORS'  and sta_range=0  GROUP BY buss_value,client_ip)a)b "+
				"where b.rn<=50";
		
		DataFrame oper_tmp = operators_id_param.sqlContext().sql(sql);
		
		JavaRDD<SAN051> oper_result_rdd = oper_tmp.toJavaRDD().map(new Function<Row, Map<String,String>>() {

			@Override
			public Map<String, String> call(Row row) throws Exception {
				Map<String, String> map = new HashMap<String, String>();
				map.put("count", row.getAs("vcount").toString());
				map.put("cip", row.getAs("client_ip").toString());
				map.put("operators_id", row.getAs("buss_value").toString());
				map.put("rank", row.getAs("rn").toString());
				return map;
			}
		}).mapPartitions(new FlatMapFunction<Iterator<Map<String,String>>, SAN051>() {

			@Override
			public Iterable<SAN051> call(Iterator<Map<String, String>> it) throws Exception {
				List<SAN051> list = new LinkedList<SAN051>();
				try{
					while(it.hasNext()){
						Map<String,String> map = it.next();
						long count = Long.parseLong(String.valueOf(map.get("count")));
						int rank = Integer.parseInt(String.valueOf(map.get("rank")));
						SAN051 sa = new SAN051();
						sa.setYear(DateUtil.getCurrentYear(date));
						sa.setHalf_year(DateUtil.getHalfYear(date));
						sa.setQuarter(DateUtil.getQuarter(date));
						sa.setMonth(DateUtil.getCurrentMonth(date));
						sa.setWeek(DateUtil.getCurrentWeek(date));
						sa.setDay(DateUtil.getCurrentDay(date));
						sa.setHour(0);
						sa.setBuss_type("WEBSITE_COUNT_OPERATORS");
						sa.setBuss_value(CommonUtils.valueOf(map.get("operators_id")));
						sa.setCount_value(count);
						sa.setSta_range(0);
						sa.setClient_ip(CommonUtils.valueOf(map.get("cip")));
						sa.setRank(rank);
						list.add(sa);
					}
				}catch(Exception e){
					SysLog.error(e.getMessage());
				}
				return list;
			}
		});
		// 生成DataFrame
		DataFrame oper_top = operators_id_param.sqlContext().createDataFrame(oper_result_rdd, SAN051.class);
		return oper_top;
	}

	@SuppressWarnings("serial")
	private DataFrame getCityTopDF(DataFrame city_param, final Date date) {
		String sql = "select b.buss_value,b.client_ip,b.vcount,b.rn from( "+
				"select a.buss_value,a.client_ip,a.vcount,ROW_NUMBER() OVER(PARTITION BY a.buss_value ORDER BY a.vcount DESC) rn from( "+
				"select buss_value,client_ip,sum(count_value)as vcount from dws_hour_san051_temp where buss_type='WEBSITE_COUNT_AREA'  and sta_range=0 GROUP BY buss_value,client_ip)a)b "+
				"where b.rn<=50";
		
		DataFrame city_tmp = city_param.sqlContext().sql(sql);
		
		JavaRDD<SAN051> cityRDD = city_tmp.toJavaRDD().map(new Function<Row, Map<String,String>>() {

			@Override
			public Map<String, String> call(Row row) throws Exception {
				Map<String, String> map = new HashMap<String, String>();
				map.put("count", row.getAs("vcount").toString());
				map.put("cip", row.getAs("client_ip").toString());
				map.put("city", row.getAs("buss_value").toString());
				map.put("rank", row.getAs("rn").toString());
				return map;
			}
		}).mapPartitions(new FlatMapFunction<Iterator<Map<String,String>>, SAN051>() {

			@Override
			public Iterable<SAN051> call(Iterator<Map<String, String>> it) throws Exception {
				List<SAN051> list = new LinkedList<SAN051>();
				try{
					while(it.hasNext()){
						Map<String,String> map = it.next();
						long count = Long.parseLong(String.valueOf(map.get("count")));
						int rank = Integer.parseInt(String.valueOf(map.get("rank")));
						SAN051 sa = new SAN051();
						sa.setYear(DateUtil.getCurrentYear(date));
						sa.setHalf_year(DateUtil.getHalfYear(date));
						sa.setQuarter(DateUtil.getQuarter(date));
						sa.setMonth(DateUtil.getCurrentMonth(date));
						sa.setWeek(DateUtil.getCurrentWeek(date));
						sa.setDay(DateUtil.getCurrentDay(date));
						sa.setHour(0);
						sa.setBuss_type("WEBSITE_COUNT_AREA");
						sa.setBuss_value(CommonUtils.valueOf(map.get("city")));
						sa.setCount_value(count);
						sa.setSta_range(0);
						sa.setClient_ip(CommonUtils.valueOf(map.get("cip")));
						sa.setRank(rank);
						list.add(sa);
					}
				}catch(Exception e){
					SysLog.error(e.getMessage());
				}
				return list;
			}
		});
		
		DataFrame df = city_param.sqlContext().createDataFrame(cityRDD, SAN051.class);
		return df;
	}

	
	
	private DataFrame testNADF(SQLContext sqlContext,
			DataFrame operators_id_param) {
		String na = "1";
		String sql = "select * from operators_id_param where OPERATORS_ID= \'"
				+ na + "\' ";
		DataFrame oper_tmp = sqlContext.sql(sql);
		return oper_tmp;
	}
}
